package cn.itcast.sql;

import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.util.Arrays;

import static org.apache.flink.table.api.Expressions.$;

/**
 * @author KTL
 * @version V1.0
 * @Package cn.itcast.sql
 * @date 2021/3/6 0006 15:06
 * @Copyright © 2015-04-29  One for each, and two for each
 *     演示Flink Table&SQL 案例- 使用SQL和Table两种方式做WordCount
 */
public class Demo2_2 {
    public static void main(String[] args) throws Exception {
        //TODO 0 env
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        final EnvironmentSettings build = EnvironmentSettings.newInstance().useBlinkPlanner().inStreamingMode().build();
        final StreamTableEnvironment tenv = StreamTableEnvironment.create(env, build);
        //TODO 1 source
        DataStreamSource<WC> wordsDS = env.fromElements(
                new WC("hello", 1),
                new WC("world", 1),
                new WC("hello", 1)
        );
        //TODO 2 thransformation
        //将DataStream转化为View或Table然后查询
        final Table table = tenv.fromDataStream(wordsDS);
        //TODO 3 sink
        final Table resultTable = table.groupBy($("word"))
                .select($("word"), $("frequency").sum().as("frequency"))
                .filter($("frequency").isEqual(2));
        final DataStream<Tuple2<Boolean, WC>> resultDS = tenv.toRetractStream(resultTable, WC.class);
        resultDS.print();
        //TODO 4 execute
        env.execute();
    }

    @Data
    @NoArgsConstructor
    @AllArgsConstructor
   public static class WC{
        public String word;
        public long frequency;
    }
}
